package com.qiandw.spark

import javax.annotation.Resource
import javax.sql.DataSource

import com.qiandw.common.cassandra.{CassandraConfig, CassandraDaoSrv}
import com.qiandw.config.DemoMysqlConfig
import org.apache.spark.sql.{Row, SparkSession}
import org.springframework.beans.factory.InitializingBean
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Component

import scala.collection.JavaConverters._

@Component
class SparkSqlSrvImpl extends SparkSqlSrv with InitializingBean {

  @Autowired
  var demoCfg: DemoMysqlConfig = _

  @Autowired
  var cassandraCfg: CassandraConfig = _

  @Resource(name = "demoDatasource")
  var demoDatasource: DataSource = _

  @Autowired
  var cassandraDao: CassandraDaoSrv = _

  @Autowired
  var scf: SparkContextFactory = _

  var ss: SparkSession = _

  val CASSANDRA_FORMAT = "org.apache.spark.sql.cassandra"
  def cassandraOptions(keyspace: String, table: String) = Map("keyspace" -> keyspace, "table" -> table)


  override def afterPropertiesSet(): Unit = {
    ss = scf.getSparkSession
    registerTables(ss)

    /**
      * 缓存dataFrame
      */
//    val df = ss.sql("select name from t_member")
//    df.cache()

    /**
      * 缓存表：创建一个缓存表
      */
//    val df = ss.sql("select * from t_member")
//    df.createTempView("t_member_cached")
//    ss.sqlContext.cacheTable("t_member_cached")
  }

  private def registerTables(ss: SparkSession): Unit = {

    cassandraDao.getTableNames.asScala.foreach(t => {
      ss.read.format(CASSANDRA_FORMAT).options(cassandraOptions(cassandraCfg.defaultKeyspace, t))
        .load().createOrReplaceTempView(t)
    })

    val jdbcRs = demoDatasource.getConnection.createStatement().executeQuery("SELECT table_name FROM information_schema.tables where table_schema='" + demoCfg.schema + "';")
    while (jdbcRs.next()) {
      val tableName = jdbcRs.getString(1)
      ss.read.format("jdbc").options(demoOptions(tableName)).load().createOrReplaceTempView(tableName)
    }

  }

  private def demoOptions(table: String): Map[String, String] = {
    Map("url" -> demoCfg.conStr,
      "driver" -> demoCfg.driverClass,
      "dbtable" -> table,
      "user" -> demoCfg.username,
      "password" -> demoCfg.password)
  }

  override def sql(sqlText: String): (Array[String], Array[Row]) = {
    val df = ss.sql(sqlText)
    (df.columns, df.collect())
  }
}
